package com.winit.cloudlink.message.internal;


import com.winit.cloudlink.common.exception.CloudlinkException;
import com.winit.cloudlink.config.Metadata;
import com.winit.cloudlink.message.Message;
import com.winit.cloudlink.message.MessageTemplate;
import com.winit.cloudlink.message.exception.MessageSendException;
import com.winit.cloudlink.message.internal.support.RetryTemplateEx;
import com.winit.cloudlink.message.utils.FileMessageWriter;
import com.winit.cloudlink.message.utils.MessageWriter;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.rabbit.support.RabbitExceptionTranslator;
import org.springframework.retry.RecoveryCallback;
import org.springframework.retry.RetryCallback;
import org.springframework.retry.RetryContext;
import org.springframework.retry.RetryListener;
import org.springframework.retry.backoff.ExponentialBackOffPolicy;
import org.springframework.retry.policy.SimpleRetryPolicy;
import org.springframework.retry.support.RetryTemplate;

import java.io.IOException;


public abstract class AbstractMessageTemplate implements MessageTemplate {
	private static final Logger logger = LoggerFactory
			.getLogger(AbstractMessageTemplate.class);
	protected Metadata metadata;
	protected MessageSender messageSender;
	protected MessageReceiver messageReceiver;
	private volatile RetryTemplate retryTemplate;
	private MessageWriter messageWriter;
	private volatile RecoveryCallback<?> recoveryCallback;

	public MessageSender getMessageSender() {
		return messageSender;
	}

	public MessageReceiver getMessageReceiver() {
		return messageReceiver;
	}

	public AbstractMessageTemplate(Metadata metadata) {
		this.metadata = metadata;
		//设置重试策略，向调用者抛出异常前，会做五次尝试）.
		RetryTemplate retryTemplate = new RetryTemplateEx();
		SimpleRetryPolicy retryPolicy = new SimpleRetryPolicy();
		retryPolicy.setMaxAttempts(5);
		retryTemplate.setRetryPolicy(retryPolicy);
		//重试回退策略采用指数退避策略，initialInterval指定初始休眠时间，设置为100毫秒，maxInterval指定最大休眠时间，设置为3秒，multiplier指定乘数，设置为3，即下一次休眠时间为当前休眠时间*multiplier
		ExponentialBackOffPolicy backOffPolicy = new ExponentialBackOffPolicy();
		backOffPolicy.setInitialInterval(100);
		backOffPolicy.setMaxInterval(10000);
		backOffPolicy.setMultiplier(3);
		retryTemplate.setBackOffPolicy(backOffPolicy);
		retryTemplate.registerListener(new RetryListener() {
			@Override
			public <T, E extends Throwable> boolean open(RetryContext context, RetryCallback<T, E> callback) {
				return true;
			}

			@Override
			public <T, E extends Throwable> void close(RetryContext context, RetryCallback<T, E> callback, Throwable throwable) {
			}

			@Override
			public <T, E extends Throwable> void onError(RetryContext context, RetryCallback<T, E> callback, Throwable throwable) {
				logger.error("Failure to send message at {} times.", context.getRetryCount(), throwable);
			}
		});
		setRetryTemplate(retryTemplate);
		//start();
		this.messageWriter = new FileMessageWriter();
	}

	public void start() {
		try {
			messageSender = buildMessageSender(metadata);

			messageReceiver = buildMessageReceiver(metadata);
		} catch (Exception e) {
			logger.error("MessageTemplate initialized error.", e);
			throw new MessageSendException("MessageTemplate initialized error.", e);
		}
	}

	public void setRetryTemplate(RetryTemplate retryTemplate) {
		this.retryTemplate = retryTemplate;
	}

	public void setRecoveryCallback(RecoveryCallback<?> recoveryCallback) {
		this.recoveryCallback = recoveryCallback;
	}


	protected MessageSender buildMessageSender(Metadata metadata) {
		MessageSender messageSender = new MessageSender();
		return messageSender;
	}

	private MessageReceiver buildMessageReceiver(Metadata metadata) {
		MessageReceiver messageReceiver = new MessageReceiver();
		return messageReceiver;
	}


	@Override
	public void send(final Message message) throws CloudlinkException {
		try {
			if (this.retryTemplate != null) {
				this.retryTemplate.execute(new RetryCallback<Void, Exception>() {
					@Override
					public Void doWithRetry(RetryContext context) throws Exception {
						doSend(message);
						return null;
					}
				}, (RecoveryCallback<Void>) this.recoveryCallback);
			} else {
				doSend(message);
			}
		} catch (Exception e) {
			//消息发送失败后，保存消息到本地
			writeMessage(message);
			if (e instanceof RuntimeException) {
				throw (RuntimeException) e;
			}
			throw RabbitExceptionTranslator.convertRabbitAccessException(e);
		}
	}

	private void doSend(Message message) throws CloudlinkException {
		try {
			preHandle(message);
			sendInternal(message);
			postHandle(message);
		} catch (Exception e) {
			String errorMsg = String.format("send message error." + message);
			logger.error(errorMsg, e);
			throw new MessageSendException("Exception occurred while sending a message after retry times", e);
		}
	}

	protected abstract void preHandle(Message message);

	protected abstract void postHandle(Message message);

	protected abstract void sendInternal(Message eventMessage);

	private void writeMessage(Message message) {
		try {
			messageWriter.write(message);
		} catch (IOException ex) {
			logger.error("write message to file error.", ex);
		}
	}
}